package study.cdc.debezium.support.embedded;

import io.debezium.engine.DebeziumEngine;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.connect.source.SourceRecord;

import java.util.List;

@Slf4j
public class SourceRecordHandler implements DebeziumEngine.ChangeConsumer<SourceRecord> {

    @Override
    public void handleBatch(List<SourceRecord> records, DebeziumEngine.RecordCommitter<SourceRecord> committer) throws InterruptedException {
        log.info(">>>>>> {}", records.size());
//
        DebeziumEngine.Offsets offsets = committer.buildOffsets();
        log.info("aaaaaaaaaaaaaaaaaaaaaaa {}", offsets);
        for (SourceRecord record : records) {
            log.info("aaaa");
            try {
                committer.markProcessed(record);
            } catch (Exception ex) {
                ex.printStackTrace();
                committer.markBatchFinished();
            }
        }
        committer.markBatchFinished();
    }
}
